package com.ss.mq.pattern.rpc;

import com.rabbitmq.client.*;
import com.ss.mq.util.RabbitUtils;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.Scanner;
import java.util.UUID;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;

/**
 * @author 叁昇
 * @create 2020-12-19-16:30
 */
public class RPCClient {
    Connection connection;
    Channel channel;

    public RPCClient() throws Exception {
        connection = RabbitUtils.getConnection();
        channel = connection.createChannel();
    }


    public static void main(String[] args) throws Exception {
        RPCClient rpcClient = new RPCClient();

        while (true) {
            System.out.print("求第几个斐波那契数: ");
            int n = new Scanner(System.in).nextInt();
            String f = rpcClient.call( "" + n );
            System.out.println( "第" + n + "个斐波拉契数为: " + f);
            System.out.println( "========================" );
        }
    }

    /**
     * 远程调用客户端的回调信息, 即获取响应数据
     * @param msg
     * @return
     */
    private String call(String msg) throws Exception {
        // 自动生成队列
        String queue = channel.queueDeclare().getQueue();

        // 随机生成关联 id
        String correlationId = UUID.randomUUID().toString();

        // 设置请求与响应的关联 id, 传递响应数据的队列
        AMQP.BasicProperties properties = new AMQP.BasicProperties().builder().correlationId(correlationId).replyTo(queue).build();

        // 向 rpc_queue 队列发送请求数据, 请求第 n 个斐波那契数
        channel.basicPublish("", "rpc_queue", properties, msg.getBytes(StandardCharsets.UTF_8));

        // 用来保存结果的阻塞集合: 取数据时, 没有数据则会阻塞暂停等待
        BlockingQueue<String> response = new ArrayBlockingQueue<>(1);

        // 接受响应数据的回调对象
        DeliverCallback deliver = (s, delivery) -> {
            String message = new String(delivery.getBody(), StandardCharsets.UTF_8);
            // 如果响应消息的关联 id 与请求的关联 id 相同, 执行处理响应数据的代码段
            if (delivery.getProperties().getCorrelationId().contentEquals(correlationId)) {
                // 把收到的响应数据存入阻塞集合
                response.offer(message);
            }
        };

        CancelCallback cancel = s -> {};

        // 开始从队列接受响应数据
        channel.basicConsume(queue, true, deliver, cancel);

        // 返回保存在集合中的响应数据
        return response.take();
    }
}
